{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7.4 数据集（Dataset）\n",
    "上一节介绍了通过队列进行多线程输入的方法。除队列以外，TensorFlow还提供了一套更高层的数据处理框架。**在新的框架中，每一个数据来源被抽象成一个“数据集”，开发者可以以数据集为基本对象，方便地进行batching、随机打乱（shuffle）等操作。从1.3版本起，TensorFlow正式推荐使用数据集作为输入数据的首选框架**。从1.4版本起，数据集框架从tf.contrib.data迁移到tf.data，成为TensorFlow的核心组成部件。\n",
    "\n",
    "### 7.4.1 数据集的基本使用方法\n",
    "**在数据集框架中，每一个数据集代表一个数据来源：数据可能来自一个张量，一个文本文件，一个TFRecord文件，或者经过sharding的一系列文件，等等。**\n",
    "\n",
    "由于训练数据通常无法全部写入内存中，从数据集中读取数据时需要使用一个法代器（iterator）按顺序进行读取，这点与队列的dequeue操作和Reader的read操作相似。与队列相似，数据集也是计算图上的一个节点。\n",
    "\n",
    "**1. 从数组创建数据集**\n",
    "\n",
    "下面先看一个简单的例子，这个例子从一个张量创建一个数据集，遍历这个数据集，并对每个输入输出$ y = x^2 $的值。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n",
      "4\n",
      "9\n",
      "25\n",
      "64\n"
     ]
    }
   ],
   "source": [
    "import tensorflow as tf\n",
    "\n",
    "# 从一个数组创建数据集\n",
    "input_data = [1, 2, 3, 5, 8]\n",
    "dataset = tf.data.Dataset.from_tensor_slices(input_data)\n",
    "\n",
    "# 定义迭代器用于便利数据。因为上面定义的数据集没有用placeholder\n",
    "# 作为输入参数，所以这里可以使用最简单的one_shot_iterator。\n",
    "iterator = dataset.make_one_shot_iterator()\n",
    "\n",
    "# get_next() 返回代表一个输入数据的张量。\n",
    "x = iterator.get_next()\n",
    "y = x * x\n",
    "\n",
    "with tf.Session() as sess:\n",
    "    for i in range(len(input_data)):\n",
    "        print(sess.run(y))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "可以看到，利用数据集读取数据有三个基本步骤：\n",
    "1. **定义数据集的构造方法**，这个例子使用`tf.data.Dataset.from_tensor_slices()`，表明数据集是从一个张量中构建的，如果数据集是文件中构建的，则需要相应调整不同的构造方法。\n",
    "2. **定义遍历器**，这个例子使用了简单的`one_shot_iterator`来遍历数据集，稍后将介绍更加灵活的`initializable_iterator`。\n",
    "3. **使用`get_next`方法从遍历器中读取数据张量，作为计算图其他部分的输入。**\n",
    "\n",
    "**2. 读取文本文件里的数据**\n",
    "\n",
    "在真实项目中，训练数据通常是保存在硬盘文件上的。比如在自然语言处理的任务中，训练数据通常是以每行一条数据的形式存在文本文件中，这时可以用`TextLineDataset`来更方便地读取数据："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "b'File1, line1.'\n",
      "b'File1, line2.'\n",
      "b'File2, line1.'\n",
      "b'File2, line2.'\n"
     ]
    }
   ],
   "source": [
    "# 创建文本文件作为本例的输入。\n",
    "with open(\"./test1.txt\", \"w\") as file:\n",
    "    file.write(\"File1, line1.\\n\") \n",
    "    file.write(\"File1, line2.\\n\")\n",
    "with open(\"./test2.txt\", \"w\") as file:\n",
    "    file.write(\"File2, line1.\\n\") \n",
    "    file.write(\"File2, line2.\\n\")\n",
    "\n",
    "# 从文本文件创建数据集。假定每行文字是一个训练例子。这里可以提供多个文件。\n",
    "input_files = [\"./test1.txt\", \"./test2.txt\"]\n",
    "dataset = tf.data.TextLineDataset(input_files)\n",
    "\n",
    "# 定义迭代器。\n",
    "iterator = dataset.make_one_shot_iterator()\n",
    "\n",
    "# 这里get_next()返回一个字符串类型的张量，代表文件中的一行。\n",
    "x = iterator.get_next()  \n",
    "with tf.Session() as sess:\n",
    "    for i in range(4):\n",
    "        print(sess.run(x))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**3. 解析TFRecord文件里的数据**\n",
    "\n",
    "在图像相关任务中，输入数据通常以TFRecord形式存储，这时可以用`TFRecordDataset`来读取数据。与文本文件不同，每一个TFRecord都有自己不同的feature格式，因此在读取TFRecord时，需要提供一个`parser`函数来解析所读取的TFRecord的数据格式。这里读取文件为本章第一节创建的文件，如下："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "7\n",
      "3\n",
      "4\n",
      "6\n",
      "1\n",
      "8\n",
      "1\n",
      "0\n",
      "9\n",
      "8\n"
     ]
    }
   ],
   "source": [
    "# 解析一个TFRecord的方法。\n",
    "def parser(record):\n",
    "    # 解析读取的一个样例\n",
    "    features = tf.parse_single_example(\n",
    "        record,\n",
    "        features={\n",
    "            'image_raw':tf.FixedLenFeature([],tf.string),\n",
    "            'pixels':tf.FixedLenFeature([],tf.int64),\n",
    "            'label':tf.FixedLenFeature([],tf.int64)\n",
    "        })\n",
    "    decoded_images = tf.decode_raw(features['image_raw'],tf.uint8)\n",
    "    retyped_images = tf.cast(decoded_images, tf.float32)\n",
    "    images = tf.reshape(retyped_images, [784])\n",
    "    labels = tf.cast(features['label'],tf.int32)\n",
    "    #pixels = tf.cast(features['pixels'],tf.int32)\n",
    "    return images, labels\n",
    "\n",
    "# 从TFRecord文件创建数据集。这里可以提供多个文件。\n",
    "input_files = [\"output.tfrecords\"]\n",
    "dataset = tf.data.TFRecordDataset(input_files)\n",
    "\n",
    "# map函数表示对数据集中的每一条数据进行调用解析方法。使用TFRecordDataset读出的\n",
    "# 是二进制的数据，这里需要通过map来调用parser对二进制数据进行解析。类似地，\n",
    "# map函数也可以用来完成其他的数据预处理工作。\n",
    "dataset = dataset.map(parser)\n",
    "\n",
    "# 定义遍历数据集的迭代器。\n",
    "iterator = dataset.make_one_shot_iterator()\n",
    "\n",
    "# 读取数据，可用于进一步计算\n",
    "image, label = iterator.get_next()\n",
    "\n",
    "with tf.Session() as sess:\n",
    "    for i in range(10):\n",
    "        x, y = sess.run([image, label]) \n",
    "        print(y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**4. 使用initializable_iterator来动态初始化数据集**\n",
    "\n",
    "以上例子使用了最简单的`one_ shot_iterator`来遍历数据集。在使用`one shot_iterator`时，数据集的所有参数必须已经确定，因此`one_shot_iterator`不需要特别的初始化过程。\n",
    "\n",
    "如果需要用placeholder来初始化数据集，那就需要用到`initializable_iterator`。以下代码给出了用`initializable_iterator`来动态初始化数据集的例子:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 从TFRecord文件创建数据集，具体文件路径是一个placeholder，稍后再提供具体路径。\n",
    "input_files = tf.placeholder(tf.string)\n",
    "dataset = tf.data.TFRecordDataset(input_files)\n",
    "dataset = dataset.map(parser)\n",
    "\n",
    "# 定义遍历dataset的initializable_iterator。\n",
    "iterator = dataset.make_initializable_iterator()\n",
    "image, label = iterator.get_next()\n",
    "\n",
    "with tf.Session() as sess:\n",
    "    # 首先初始化iterator，并给出input_files的值。\n",
    "    sess.run(iterator.initializer,\n",
    "             feed_dict={input_files: [\"output.tfrecords\"]})\n",
    "    # 遍历所有数据一个epoch。当遍历结束时，程序会抛出OutOfRangeError。\n",
    "    while True:\n",
    "        try:\n",
    "            x, y = sess.run([image, label])\n",
    "        except tf.errors.OutOfRangeError:\n",
    "            break "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**在上面的例子中，文件路径使用placeholder和feed_dict的方式传给数据集。使用这种方法，在实际项目中就不需要总是将参数写入计算图的定义，而可以使用程序参数的方式动态指定参数。**\n",
    "\n",
    "另外注意到，上面例子中的循环体不是指定循环运行10次sess.run，而是使用*while(True) - try - except*的形式来将所有数据遍历一遍（即一个epoch）。这是因为在动态指定输入数据时，不同数据来源的数据量大小难以预知，而这个方法使我们不必提前知道数据量的精确大小。\n",
    "\n",
    "以上介绍的两种iterator足以满足大多数项目的需求。除这两种以外，TensorFlow还提供了下面两种更加灵活的迭代器，这里不再具体介绍:\n",
    "- `reinitializable_iterator`可以多次initialize用于遍历不同的数据来源；\n",
    "- `feedable_iterator`可以用feed_dict的方式动态指定运行哪个iterator。\n",
    "\n",
    "### 7.4.2 数据集的高层操作\n",
    "在上一小节中介绍了数据集的基础用法。在这一小节中，将介绍数据集框架提供的一些方便实用的高层API。\n",
    "\n",
    "**1. dataset.map**\n",
    "\n",
    "在7.4.1小节中介绍过map方法来对TFRecord进行解析操作：`dataset = dataset.map(parser)`。map是在数据集上进行操作的最常用的方法之一。在这里，map(parser）方法表示对数据集中的每一条数据调用参数中指定的parser方法。**对每一条数据进行处理后，map将处理后的数据包装成一个新的数据集返回。map 函数非常灵活，可以用于对数据的任何预处理操作**。例如在前面小节中曾使用如下方法来对数据进行预处理：\n",
    "\n",
    "`distorted_image = preprocess_for_train(decoded_image, image_size, image_size, None)`\n",
    "\n",
    "而在数据集框架中，可以通过map来对每条数据调用`preprocess_for_train`方法：\n",
    "\n",
    "`dataset = dataset.map(lambda x:preprocess_for_train(x, image_size, image_size, None))`\n",
    "\n",
    "在上面的代码中，**lambda表达式的作用是将原来有4个参数的函数转化为只有1个参数的函数。**`preprocess_for_train`函数的第一个参数decoded_image变成了lambda表达式中的x，这个参数就是原来函数中的参数decoded_image.preprocess_for_train函数中后3个参数都被换成了具体的数值。注意这里的image_size是一个变量，有具体取值，该值需要在程序的上文中给出。\n",
    "\n",
    "**从表面上看，新的代码在长度上似乎并没有缩短，然而由于map方法返回的是一个新的数据集，可以直接继续调用其他高层操作。在上一节介绍的队列框架中，预处理、shuffle、batch等操作有的在队列上进行，有的在图片张量上进行，整个处理流程在处理队列和张量的代码片段中来回切换。而在数据集操作中，所有操作都在数据集上进行，这样的代码结构将非常的干净、简洁。**\n",
    "\n",
    "**2. dataset.batch & dataset.shuffle**\n",
    "\n",
    "类似于队列框架中的`tf.train.batch`和`tf.train.shuffle_batch`，这两个操作在数据集框架中的实现如下：\n",
    "\n",
    "- `dataset = dataset.batch(batch_size)`，**将数据组合成batch，参数batch_size代表要输出的每个batch由多少条数据组成。如果数据集中包含多个张量，那么batch操作将对每一个张量分开进行**。举例而言，如果数据集中的每一个数据（即iterator.get_next()的返回值）是image、label两个张量，其中image的维度是[300, 300]，label的维度是[]，batch_size是128，那么经过batch 操作后的数据集的每一个输出将包含两个维度分别是[128, 300, 300]和[128]的张量。\n",
    "\n",
    "- `dataset = dataset.shuffle(buffer_size)`，随机打乱顺序，参数buffer_size等效于`tf.train.shuffle_batch`中的参数min_after_dequeue。shuffle算法在内部使用一个缓冲区中保存buffer_size条数据，每读入一条新数据时，从这个缓冲区中随机选择一条数据进行输出。缓冲区的大小越大，随机的性能越好，但占用的内存也越多。\n",
    "\n",
    "**3. dataset.repeat**\n",
    "\n",
    "repeat是另一个常用的操作方法，它将数据集中的数据复制多份，其中每一份被称为一个epoch。\n",
    "\n",
    "`dataset = dataset.repeat(N)`\n",
    "\n",
    "上面代码将数据重复N份。需要指出的是，**如果数据集在repeat前己经进行了shuffle操作，输出的每个epoch中随机shuffle的结果并不会相同。**例如，如果输入数据是[1, 2, 3]，shuffle后输出的第一个epoch是[2, 1, 3]，而第二个epoch则有可能是[3, 2, 1]。repeat和map、shuffle、batch等操作一样，都只是计算图中的一个计算节点。repeat只代表重复相同的处理过程，并不会记录前一个epoch的处理结果。\n",
    "\n",
    "**4. 其他数据集操作**\n",
    "\n",
    "除这些方法以外，数据集还提供了其他多种操作。例如:\n",
    "- concatenate()将两个数据集顺序连接起来;\n",
    "- take(N)从数据集中读取前N项数据;\n",
    "- skip(N)在数据集中跳过前N项数据;\n",
    "- flap_map()从多个数据集中轮流读取数据;\n",
    "\n",
    "等等，这里不再一一介绍，有需要的读者可以查询TensorFlow相关文档。\n",
    "\n",
    "以下例子将这些方法组合起来，使用数据集实现了7.3.4小节中的数据输入流程。与7.3.4小节中介绍的类似，该例子从文件中读取原始数据，进行预处理、shuffle、batching等操作，并通过repeat方法训练多个epoch。不同的是，以下例子在训练数据集之外，还另外读取了测试数据集，并对测试集和数据集进行了略微不同的预处理。在训练时，调用7.2.2小节中的`preprocess_for_train`方法对图像进行随机反转等预处理操作；而在测试时，测试数据以原本的样子直接输入测试。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test accuracy is: 0.8962\n"
     ]
    }
   ],
   "source": [
    "import tensorflow as tf\n",
    "\n",
    "# 1. 列举输入文件。训练和测试使用不同的数据\n",
    "train_files = tf.train.match_filenames_once(\"output.tfrecords\")\n",
    "test_files = tf.train.match_filenames_once(\"output_test.tfrecords\")\n",
    "\n",
    "\n",
    "# 2. 解析一个TFRecord的方法。\n",
    "def parser(record):\n",
    "    features = tf.parse_single_example(\n",
    "        record,\n",
    "        features={\n",
    "            'image_raw':tf.FixedLenFeature([],tf.string),\n",
    "            'pixels':tf.FixedLenFeature([],tf.int64),\n",
    "            'label':tf.FixedLenFeature([],tf.int64)\n",
    "        })\n",
    "    decoded_images = tf.decode_raw(features['image_raw'],tf.uint8)\n",
    "    retyped_images = tf.cast(decoded_images, tf.float32)\n",
    "    images = tf.reshape(retyped_images, [784])\n",
    "    labels = tf.cast(features['label'],tf.int32)\n",
    "    #pixels = tf.cast(features['pixels'],tf.int32)\n",
    "    return images, labels\n",
    "\n",
    "\n",
    "# 3. 定义训练数据集\n",
    "image_size = 299          # 定义神经网络输入层图片的大小。\n",
    "batch_size = 100          # 定义组合数据batch的大小。\n",
    "shuffle_buffer = 10000   # 定义随机打乱数据时buffer的大小。\n",
    "\n",
    "# 定义读取训练数据的数据集。\n",
    "dataset = tf.data.TFRecordDataset(train_files)\n",
    "dataset = dataset.map(parser)\n",
    "\n",
    "# 对数据集进行预处理，这里略去\n",
    "# dataset = dataset.map(lambda image, label:(\n",
    "#                          preprocess_for_train(image, image_size, image_size, None), label)\n",
    "\n",
    "# 对数据进行shuffle和batching操作。这里省略了对图像做随机调整的预处理步骤。\n",
    "dataset = dataset.shuffle(shuffle_buffer).batch(batch_size)\n",
    "\n",
    "# 重复NUM_EPOCHS个epoch。间接指定训练的轮数\n",
    "NUM_EPOCHS = 10\n",
    "dataset = dataset.repeat(NUM_EPOCHS)\n",
    "\n",
    "# 定义数据集迭代器。虽然定义数据集时没有直接使用placeholder来提供文件地址，但是\n",
    "# tf.train.match_filenames_once方法得到的结果和与placeholder的机制类似，也需要\n",
    "# 初始化，所以这里使用的是initializable_iterator。\n",
    "iterator = dataset.make_initializable_iterator()\n",
    "image_batch, label_batch = iterator.get_next()\n",
    "\n",
    "\n",
    "# 4. 定义神经网络的结构个优化过程。这里与7.3.4小节相同。\n",
    "def inference(input_tensor, weights1, biases1, weights2, biases2):\n",
    "    layer1 = tf.nn.relu(tf.matmul(input_tensor, weights1) + biases1)\n",
    "    return tf.matmul(layer1, weights2) + biases2\n",
    "\n",
    "INPUT_NODE = 784\n",
    "OUTPUT_NODE = 10\n",
    "LAYER1_NODE = 500\n",
    "REGULARAZTION_RATE = 0.0001   \n",
    "TRAINING_STEPS = 5000        \n",
    "\n",
    "weights1 = tf.Variable(tf.truncated_normal([INPUT_NODE, LAYER1_NODE], stddev=0.1))\n",
    "biases1 = tf.Variable(tf.constant(0.1, shape=[LAYER1_NODE]))\n",
    "\n",
    "weights2 = tf.Variable(tf.truncated_normal([LAYER1_NODE, OUTPUT_NODE], stddev=0.1))\n",
    "biases2 = tf.Variable(tf.constant(0.1, shape=[OUTPUT_NODE]))\n",
    "\n",
    "y = inference(image_batch, weights1, biases1, weights2, biases2)\n",
    "    \n",
    "# 计算交叉熵及其平均值\n",
    "cross_entropy = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=y, labels=label_batch)\n",
    "cross_entropy_mean = tf.reduce_mean(cross_entropy)\n",
    "    \n",
    "# 损失函数的计算\n",
    "regularizer = tf.contrib.layers.l2_regularizer(REGULARAZTION_RATE)\n",
    "regularaztion = regularizer(weights1) + regularizer(weights2)\n",
    "loss = cross_entropy_mean + regularaztion\n",
    "\n",
    "# 优化损失函数\n",
    "train_step = tf.train.GradientDescentOptimizer(0.01).minimize(loss)\n",
    "\n",
    "\n",
    "# 5. 定义测试用的数据集及相关\n",
    "# 与训练时不同，测试数据的Dataset不需要经过随机翻转等预处理操作，\n",
    "# 也不需要打乱顺序和重复多个epoch，这里使用与训练数据相同的parser\n",
    "# 进行解析，调整分辨率到网络输入层大小，然后直接进行batching操作。\n",
    "# 定义测试用的Dataset。\n",
    "test_dataset = tf.data.TFRecordDataset(test_files)\n",
    "test_dataset = test_dataset.map(parser)\n",
    "test_dataset = test_dataset.batch(batch_size)\n",
    "\n",
    "# 定义测试数据上的迭代器。\n",
    "test_iterator = test_dataset.make_initializable_iterator()\n",
    "test_image_batch, test_label_batch = test_iterator.get_next()\n",
    "\n",
    "# 定义测试数据上的预测结果为logits值最大的分类。\n",
    "test_logit = inference(test_image_batch, weights1, biases1, weights2, biases2)\n",
    "predictions = tf.argmax(test_logit, axis=-1, output_type=tf.int32)\n",
    "\n",
    "\n",
    "# 6. 声明会话并运行神经网络的优化过程。\n",
    "with tf.Session() as sess:  \n",
    "    # 初始化变量。\n",
    "    sess.run((tf.global_variables_initializer(),\n",
    "              tf.local_variables_initializer()))\n",
    "    \n",
    "    # 初始化训练数据的迭代器。\n",
    "    sess.run(iterator.initializer)\n",
    "    \n",
    "    # 循环进行训练，直到数据集完成输入、抛出OutOfRangeError错误。\n",
    "    while True:\n",
    "        try:\n",
    "            sess.run(train_step)\n",
    "        except tf.errors.OutOfRangeError:\n",
    "            break\n",
    "\n",
    "    # 初始化测试数据的迭代器。\n",
    "    sess.run(test_iterator.initializer)\n",
    "    # 获取预测结果。\n",
    "    test_results = []\n",
    "    test_labels = []\n",
    "    while True:\n",
    "        try:\n",
    "            pred, label = sess.run([predictions, test_label_batch])\n",
    "            test_results.extend(pred)\n",
    "            test_labels.extend(label)\n",
    "        except tf.errors.OutOfRangeError:\n",
    "            break\n",
    "\n",
    "# 计算准确率\n",
    "correct = [float(y == y_) for (y, y_) in zip (test_results, test_labels)]\n",
    "accuracy = sum(correct) / len(correct)\n",
    "print(\"Test accuracy is:\", accuracy) "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
